package org.xiaoleng.cdm.connections;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.xiaoleng.cdm.common.CdmConstants;
import org.xiaoleng.cdm.convertors.MysqlFieldConvertor;
import org.xiaoleng.cdm.datasources.Datasource;
import org.xiaoleng.cdm.datasources.MysqlDatasource;
import org.xiaoleng.cdm.divers.DriverLoader;

import java.sql.*;
import java.util.*;
import java.util.function.Function;

@Getter
@Setter
@Slf4j
public class MySqlConnection extends DatabaseConnection{

    private String name;

    private Connection connection;

    private MysqlDatasource mysqlDatasource;

    private Map<String, String> fieldsMap;

    public MySqlConnection(Datasource datasource) {
        super(datasource);
        try {
            mysqlDatasource = (MysqlDatasource) datasource;
            DriverLoader.load(mysqlDatasource.getDriverPath(), mysqlDatasource.getDriverName());
            connection = DriverManager.getConnection(mysqlDatasource.getUrl(), mysqlDatasource.getUserName(), mysqlDatasource.getPassword());
        } catch (SQLException e) {
            log.error("");
            throw new RuntimeException(e);
        }
    }

    @Override
    public int read(Function<List<JSONObject>, Integer> function) {
        int count = 0;
        List<JSONObject> jsons = new ArrayList<>();
        try {
            PreparedStatement preparedStatement = connection.prepareStatement(mysqlDatasource.getSql(),
                ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            preparedStatement.setFetchSize(20);
            ResultSet resultSet = preparedStatement.executeQuery();
            while (resultSet.next()) {
                JSONObject jsonObject = new JSONObject();
                jsons.add(jsonObject);
                if (jsons.size() == 10) {
                    function.apply(jsons);
                    jsons.clear();
                }
                count++;

            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        if (!jsons.isEmpty()) {
            function.apply(jsons);
            jsons.clear();
        }
        return count;
    }

    @Override
    public int write(List<JSONObject> jsons) {
        int count = 0;
        PreparedStatement preparedStatement = null;
        Map<String, String> fields = mysqlDatasource.getFields();
        try {
            preparedStatement = connection.prepareStatement(getSqlPattern());
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        List<List<JSONObject>> partitions = Lists.partition(jsons, CdmConstants.BATCH_SIZE);
        for (List<JSONObject> partition : partitions) {
            for (JSONObject jsonObject : partition) {
                int index = 1;
                for (String field : fields.keySet()) {
                    try {
                        if (jsonObject.get(field) == null) {
                            preparedStatement.setObject(index, null);
                        }
                        preparedStatement.setObject(index, MysqlFieldConvertor.convert(fields.get(field), String.valueOf(jsonObject.get(field))));
                        index++;
                    } catch (SQLException e) {
                        throw new RuntimeException(e);
                    }
                }
                try {
                    preparedStatement.addBatch();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
            try {
                preparedStatement.executeBatch();
                preparedStatement.clearBatch();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return count;
    }

    private String getSqlPattern() {
        List<String> fields = mysqlDatasource.getFields().keySet().stream().toList();
        return String.format("INSERT INTO `%s`.`%s`(`%s`) VALUES (?, ?);", mysqlDatasource.getDatabase(),
            mysqlDatasource.getTableName(), String.join("`, `", fields));
    }

    @Override
    public void close() {
        if (connection != null) {
            try {
                connection.close();
                log.error("connection close successfully.");
            } catch (SQLException e) {
                log.error("connection close failed.");
                throw new RuntimeException(e);
            }
        }
    }

    private Map<String, String> getTableFields() {
        if (fieldsMap != null) {
            return fieldsMap;
        }
        try {
            DatabaseMetaData metaData = connection.getMetaData();
            ResultSet columns = metaData.getColumns(null, null, mysqlDatasource.getTableName(), "%");
            while (columns.next()) {
                String columnName = columns.getString("COLUMN_NAME");
                String columnType = columns.getString("TYPE_NAME");
                fieldsMap.put(columnName, columnType);
            }

        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        return fieldsMap;
    }

}
